package com.tjcd.project.library_base.common.utils;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

/**
 * @Author jiangKunKun
 * @CreateDate 2025/3/3
 * @Description 消息队列
 * @SpittingGrooves
 */
public class RxDataQueueManager<T> {
    private final BlockingQueue<T> messageQueue = new LinkedBlockingQueue<>();
    private BiConsumer<T, Runnable> messageSender;

    /**
     * 构造函数接受一个BiConsumer，它接收一条消息和一个Runnable。
     * 这个Runnable可以在外部定义如何处理这条消息。
     */
    public RxDataQueueManager(BiConsumer<T, Runnable> messageSender) {
        this.messageSender = messageSender;
    }

    /**
     * 提供一个方法用于外部添加消息到队列。
     */
    public void addMessage(T message) {
        if (message != null) { // 检查消息是否为null
            try {
                messageQueue.put(message); // 添加消息到队列，如果队列已满会等待
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断状态
                System.err.println("Failed to add message due to interruption.");
            }
        }
    }

    /**
     * 启动消息处理流程，按100毫秒的间隔有序发送消息，并返回一个Disposable以便于管理生命周期。
     */
    public Disposable startMessageProcessing() {
        return Observable.interval(100, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .subscribe(l -> {
                    T message = messageQueue.poll(); // 尝试获取并移除队列的头，不会阻塞
                    if (message != null) {
                        send(message);
                    }
                    // 如果 message 为 null，则表示当前没有消息需要发送，直接跳过
                });
    }

    /**
     * 发送消息的方法，这里调用外部提供的BiConsumer来处理消息。
     */
    private void send(T message) {
        Runnable runnable = () -> {
            // 默认行为，可以根据需要在外部逻辑中覆盖
        };
        messageSender.accept(message, runnable);
    }

    public static void main(String[] args) throws InterruptedException {
        // 定义如何发送消息的逻辑
        BiConsumer<String, Runnable> sender = (message, runnable) -> {
            System.out.println("Preparing to send: " + message);

            // 版本检查等逻辑
            if (System.getProperty("java.version").compareTo("1.8") >= 0) {
                System.out.println("Handling message for Java 8 and above: " + message);
            } else {
                System.out.println("Handling message for Java versions below 8: " + message);
            }

            // 执行默认行为或其他额外处理
            runnable.run();
        };
        RxDataQueueManager<String> handler = new RxDataQueueManager<>(sender);
        Disposable disposable = handler.startMessageProcessing();
        // 示例：向消息队列添加一些数据
        for (int i = 0; i < 5; i++) {
            handler.addMessage("Message-" + i);
            Thread.sleep(50); // 模拟不同时间点添加消息
        }
        // 给足够的时间让所有消息被处理
        Thread.sleep(1000);
        // 停止消息处理
        disposable.dispose();
    }
}
